package org.websocket.service;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.websocket.dto.WebSocketBuffer;
import org.websocket.dto.WebSocketUser;
import org.websocket.encoder.WebSocketEncoder;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
@ServerEndpoint(value = "/online/{token}", encoders = {WebSocketEncoder.class})
public class WebSocketServer implements Authorization<WebSocketUser> {

    /**
     * 服务器在线人数
     */
    protected static final AtomicInteger ONLINE_QUANTITY = new AtomicInteger();

    /**
     * 服务器在线用户缓存
     */
    protected static final Map<String, Session> SESSION_CACHE = new ConcurrentHashMap<>();

    /**
     * 服务器缓存消息信息
     * <p>
     * { 'obtainId' : { 'pusherId':[] } }
     * </p>
     */
    protected static final Map<String, Map<String, Set<WebSocketBuffer>>> MESSAGE_CACHE = new ConcurrentHashMap<>();

    @Override
    public WebSocketUser auth(String token) {
        return WebSocketUser.builder().userId(token).userName("test").build();
    }

    /**
     * 连接服务
     *
     * @param session
     * @param token
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "token") String token) throws IOException {

        //获取用户信息
        WebSocketUser user = this.auth(token);

        //成功建立连接后加入
        SESSION_CACHE.putIfAbsent(user.getUserId(), session);

        //当前在线数量+1
        ONLINE_QUANTITY.incrementAndGet();
        log.info("websocket:" + user.getUserId() + "加入连接,当前在线用户" + ONLINE_QUANTITY +
                "未读消息数：" + getMessageCount(user.getUserId()));

        //推送离线消息
        this.pushCacheMessage(user.getUserId());
    }

    /**
     * 消息推送
     *
     * @param message
     * @throws IOException
     */
    @OnMessage
    public void onMessage(String message) {
        WebSocketBuffer webSocketBuffer = JSON.parseObject(message, WebSocketBuffer.class);
        log.info("websocket:" + webSocketBuffer.getObtainId() + "收到，来自" + webSocketBuffer.getPusherId()
                + "发送的消息" + webSocketBuffer.getMessage());
        this.pushRealTimeMessages(webSocketBuffer);
    }

    /**
     * 服务关闭
     *
     * @param closeReason
     * @param token
     */
    @OnClose
    public void onClose(CloseReason closeReason, @PathParam(value = "token") String token) {

        //获取用户信息
        WebSocketUser user = this.auth(token);

        //移除当前用户
        SESSION_CACHE.remove(user.getUserId());

        //在线人数减一
        ONLINE_QUANTITY.decrementAndGet();

        //打印关闭信息
        log.info("websocket:" + user.getUserId() + "断开连接,当前在线用户" + ONLINE_QUANTITY
                + ",断开原因" + closeReason.getReasonPhrase());
    }

    /**
     * 发送错误
     *
     * @param throwable
     */
    @OnError
    public void onError(Throwable throwable) {
        log.error("websocket: 发生了错误");
        throwable.printStackTrace();
    }

    /**
     * 判断用户是否在线
     *
     * @return boolean
     */
    public boolean survival(String receiverId) {
        boolean flag = !ObjectUtils.isEmpty(SESSION_CACHE.get(receiverId));
        log.info("websocket: " + receiverId + ":" + (flag ? "在线" : "离线"));
        return flag;
    }

    /**
     * 推送实时消息
     *
     * @return void
     */
    public void pushRealTimeMessages(WebSocketBuffer webSocketBuffer) {
        try {
            switch (webSocketBuffer.getMessageEnum()) {

                // 广播消息
                case fanout:

                    SESSION_CACHE.values().forEach(session -> {
                        try {
                            session.getBasicRemote().sendObject(webSocketBuffer);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    });

                    // 离线
                    this.storeCacheMessage(webSocketBuffer);

                    log.info("websocket: 广播消息" + webSocketBuffer);
                    break;

                // 私发消息
                case direct:

                    if (this.survival(webSocketBuffer.getObtainId())) {

                        // 在线
                        Session session = SESSION_CACHE.get(webSocketBuffer.getObtainId());
                        session.getBasicRemote().sendObject(webSocketBuffer);
                    } else {

                        // 离线
                        this.storeCacheMessage(webSocketBuffer);
                    }

                    log.info("websocket: 私发消息" + webSocketBuffer);
                    break;

            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("websocket: 发送消息发生了错误");
        }
    }

    /**
     * 用户离线时把消息存储到内存
     *
     * @return void
     */
    public void storeCacheMessage(WebSocketBuffer webSocketBuffer) {

        //消息接收者有未收消息
        if (MESSAGE_CACHE.containsKey(webSocketBuffer.getObtainId())) {
            Map<String, Set<WebSocketBuffer>> relations = MESSAGE_CACHE.get(webSocketBuffer.getObtainId());
            Set<WebSocketBuffer> messages = new ConcurrentSkipListSet<>(
                    Comparator.comparing(WebSocketBuffer::getTimestamp));

            // 消息发送者给消息接收者发送过消息
            if (relations.containsKey(webSocketBuffer.getPusherId())) {
                messages = relations.get(webSocketBuffer.getPusherId());
            }
            messages.add(webSocketBuffer);
            relations.putIfAbsent(webSocketBuffer.getPusherId(), messages);
            MESSAGE_CACHE.putIfAbsent(webSocketBuffer.getObtainId(), relations);
        } else {

            //消息接收者无未收消息
            Set<WebSocketBuffer> messages = new ConcurrentSkipListSet<>(
                    Comparator.comparing(WebSocketBuffer::getTimestamp));
            messages.add(webSocketBuffer);
            Map<String, Set<WebSocketBuffer>> relations = new ConcurrentHashMap<>();
            relations.putIfAbsent(webSocketBuffer.getPusherId(), messages);
            MESSAGE_CACHE.putIfAbsent(webSocketBuffer.getObtainId(), relations);
        }
    }

    /**
     * 用户上线时推送离线消息
     *
     * @return void
     */
    public void pushCacheMessage(String receiverId) throws IOException {

        //用户没有离线消息
        if (!MESSAGE_CACHE.containsKey(receiverId)) {
            return;
        }

        //用户没有离线消息
        if (ObjectUtils.isEmpty(MESSAGE_CACHE.get(receiverId))) {
            return;
        }

        //用户存在未收消息
        Map<String, Set<WebSocketBuffer>> relations = MESSAGE_CACHE.get(receiverId);
        for (Set<WebSocketBuffer> messages : relations.values()) {
            if (messages != null && messages.size() > 0) {
                Iterator<WebSocketBuffer> message = messages.iterator();
                while (message.hasNext()) {
                    WebSocketBuffer webSocketBuffer = message.next();
                    this.onMessage(JSON.toJSONString(webSocketBuffer));
                    messages.remove(webSocketBuffer);
                }
            }
        }
    }

    /**
     * 获取该用户未读的消息数量
     */
    public int getMessageCount(String recvUserId) {
        //获取该用户所有未收的消息
        Map<String, Set<WebSocketBuffer>> relations = MESSAGE_CACHE.get(recvUserId);
        if (relations != null) {
            Set<WebSocketBuffer> list = relations.get(recvUserId);
            if (list != null) {
                return relations.get(recvUserId).size();
            } else {
                return 0;
            }
        } else {
            return 0;
        }
    }
}
